/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package dpl.sender;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
//import java.io.InputStream;
//import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

/**
 *
 * @author Li Mingyang
 */
public class DplSender {

    public static void main(String[] args) throws RESessionException, REConnectionException, FileNotFoundException, IOException {
        
        List<byte[]> picBytes=new ArrayList<>();
        
        File pics=new File("D:\\lmy\\dpl-sender\\pic");
        String[] paths=pics.list();
        for(String path:paths){
//        System.out.println(path);
            File pic=new File("D:\\lmy\\dpl-sender\\pic\\"+path);
        FileInputStream is=new FileInputStream(pic);
//        BufferedImage bi=back
        byte[] buf=new byte[(int)pic.length()];
//        byte[] buf=new byte[9000];
        int num=is.read(buf);
            System.out.println(num);
        is.close();
        picBytes.add(buf);
                }
        System.out.println("picBytes.size   is "+picBytes.size());
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定
        REConnection conn = new REConnection("10.2.8.11:9876");

        //生成builder，参数为topic，由具体业务决定
        RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder("mm_mq");
        //设置数据结构，参数为每一列的列名、列类型、是否可以为空。注意： 如果设置为不能为空，那么如果添加数据时没有添加该列数据，会报错。
        builder.addColumn("time", REFieldType.Long, true);
        builder.addColumn("value", REFieldType.String, true);
        builder.addColumn("pic", REFieldType.Binary, true); 
        builder.addColumn("name", REFieldType.String, true);

        //生成session
        RESendSession session = (RESendSession) builder.build();
        session.start();
        try {
            //开始发送。
            //这里，DocSchemaName和UserDesc非必须，具体什么时候需要，由具体业务决定。
            //DocSchemaName设置一次后不会清空，每次flush都是同一个，如果需要改变，需要再次设置。
            session.setDocSchemaName("ggcc.testTable");
            //UserDesc设置后，只会在一次flush起作用，下一批数据发送需要重新设置。
            HashMap<String, String> user_desc = new HashMap<String, String>();
            user_desc.put("user", "test");
            user_desc.put("pwd", "pwd");
            user_desc.put("round", "all");
            for (int r = 0; r < 10000; ++r) {
                session.setUserDesc(user_desc);
                //添加数据。set完所有列后add即可添加一条数据。多条数据打一包发送。
                //各种set方法可以通过session.set补全来查看。set方法和REFieldType各类对应。  
                //注意：尽量不要一条数据发送一次，这样会带来较大的额外开销
                //注意2：本地缓存数据量有大小限制，如果超过500M，会报错。
                //注意3：如果要用set（index，value）的方式设置数据，index从1开始。
                try {
                    Thread.sleep(850);
                } catch (InterruptedException ex) {
                }
                for (int i = 0; i < 1; ++i) {
                    session.setLong("time", System.currentTimeMillis() / 1000L);
                    session.setString("name", "valuewb" + i);
//                    session.setBinary("pic", new StringBuilder().append("test test").toString().getBytes());
//                    ByteBuffer bb=ByteBuffer.wrap(picBytes.get(i));
//                     session.setBinary("pic",bb.array());
                    session.setBinary("pic",picBytes.get(i));
                    session.setString("value", i +"value");
                    session.add();
                    if (i % 3000 == 0) {
                        //获得各种统计量
                        //当前缓存的数据总量
                        System.out.println("current data size : " + session.getCurrentDataSize());
                        //当缓存数据较多时，数据会被分割为小包，查看当前缓存的数据包数
                        System.out.println("current pac num : " + session.getCurrentPacNum());
                        //查看当前数据包中数据的条数
                        System.out.println("record num int current pac : " + session.getRecordNumInCurrentPac());
                    }
                }
                System.out.println("current data size : " + session.getCurrentDataSize());
                System.out.println("current pac num : " + session.getCurrentPacNum());
                System.out.println("record num int current pac : " + session.getRecordNumInCurrentPac());
                System.out.println(new Date().toLocaleString());
                //添加完所有数据，flush即可发送。
                session.flush();
            }
        } finally {
            //session结束
            session.shutdown();
        }
//    }
    }

}
